package inesv.utils;

import java.util.Properties;

import com.aliyun.openservices.ons.api.*;

/**
 * RocketMQU工具类
 * 
 * @author Administrator
 *
 */
public class RocketMQUtil {
	// 您在MQ控制台创建的Producer ID
	private final static String ProducerId = "PID_inesv_demo0";
	// 您在MQ控制台创建的Producer ID
	private final static String ConsumerId = "CID_inesv_demo0";
	// 鉴权用AccessKey，在阿里云服务器管理控制台创建
	private final static String AccessKey = "LTAIXpyIpJfwSYu9";
	// 鉴权用SecretKey，在阿里云服务器管理控制台创建
	private final static String SecretKey = "yUjslKfQrz2BF2nmwI25Zjlqj8FOev";
	// 设置 TCP 接入域名（此处以公共云的公网接入为例）
	private final static String ONSAddr = "http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet";
	// 在控制台创建的Topic，即该消息所属的Topic名称
	private final static String ToPic = "inesv_demo0";

	/**
	 * 生产消息
	 * 
	 * @param tag
	 *            ：Message Tag, 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
	 * @param body
	 *            ： Message Body ，任何二进制形式的数据， MQ不做任何干预，
	 *            需要Producer与Consumer协商好一致的序列化和反序列化方式
	 * @param msgKey
	 *            ： 设置代表消息的业务关键属性，请尽可能全局唯一，以方便您在无法正常收到消息情况下，可通过MQ控制台查询消息并补发，
	 *            注意：不设置也不会影响消息正常收发
	 * @return
	 */
	public static String msgProducer(String tag, String body, String msgKey) {
		String msgId = "";
		SendResult sendResult = null;
		Properties properties = new Properties();

		properties.put(PropertyKeyConst.ProducerId, ProducerId);

		properties.put(PropertyKeyConst.AccessKey, AccessKey);

		properties.put(PropertyKeyConst.SecretKey, SecretKey);

		properties.put(PropertyKeyConst.ONSAddr, ONSAddr);
		Producer producer = ONSFactory.createProducer(properties);
		// 在发送消息前，必须调用start方法来启动Producer，只需调用一次即可
		producer.start();

		Message msg = new Message(ToPic, tag, body.getBytes());
		msg.setKey(msgKey);
		// 发送消息，只要不抛异常就是成功
		try {
			sendResult = producer.send(msg);
			msgId = sendResult.getMessageId();
		} catch (Exception e) {
			System.out.println("MQ error--->" + e.getMessage());
			return msgId;
		}
		System.out.println("Send Message success. Message ID is: " + sendResult.getMessageId());
		return msgId;
	}

	public static void msgComsumer(String topic ) {
     
		Properties properties = new Properties();
		// 您在MQ控制台创建的Consumer ID
		properties.put(PropertyKeyConst.ConsumerId, ConsumerId);
		// 鉴权用AccessKey，在阿里云服务器管理控制台创建
		properties.put(PropertyKeyConst.AccessKey, AccessKey);
		// 鉴权用SecretKey，在阿里云服务器管理控制台创建
		properties.put(PropertyKeyConst.SecretKey, SecretKey);
		// 设置 TCP 接入域名（此处以公共云公网环境接入为例）
		properties.put(PropertyKeyConst.ONSAddr, ONSAddr);
		Consumer consumer = ONSFactory.createConsumer(properties);
		consumer.subscribe(topic, "*", new MessageListener() {
			public Action consume(Message message, ConsumeContext context) {
				 
				System.out.println("--------------Receive: " + message);
				return Action.CommitMessage;
			}
		});
		consumer.start();
		System.out.println("Consumer Started");
		 
	}

	public static void main(String[] args) {
		/*String res = RocketMQUtil.msgProducer("lin_tag", "hello lqh!!!", "lin_001");
		System.out.println("res:" + res);*/

		
		  RocketMQUtil.msgComsumer("inesv_demo0");
		 
	}

}
